--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 3fd514fd4989a94585e6b9cf261410aa8e9ff58c
Parents : 774c3a2
Author : Ivan <ivan@quad4.io>
Signature : Invalid signer <e46112d44649266d71fe2193e00a4710>, author is <ivan@quad4.io>
Date : 2026-07-07T23:36:13-05:00
feat(meshchat): add input coercion for integer configuration values to improve error handling in update_config
Changes
4 files changed, 69 insertions(+), 179 deletions(-)
Diff
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index 2fdc7e28..9ef57e4d 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -14916,6 +14916,16 @@ class ReticulumMeshChat:
return value.lower() == "true"
return bool(value)
+ # Coerce untrusted config input to int. Returns None when the value
+ # cannot be converted so update_config can skip or fall back instead of
+ # raising on None/strings/lists received via the websocket API.
+ @staticmethod
+ def _coerce_int(value):
+ try:
+ return int(value)
+ except (TypeError, ValueError):
+ return None
+
async def update_config(self, data):
# update display name in config
if "display_name" in data and data["display_name"] != "":
@@ -14934,10 +14944,16 @@ class ReticulumMeshChat:
# update auto announce interval
if "auto_announce_interval_seconds" in data:
# auto auto announce interval
- auto_announce_interval_seconds = int(data["auto_announce_interval_seconds"])
- self.config.auto_announce_interval_seconds.set(
+ auto_announce_interval_seconds = self._coerce_int(
data["auto_announce_interval_seconds"],
)
+ if auto_announce_interval_seconds is None:
+ auto_announce_interval_seconds = (
+ self.config.auto_announce_interval_seconds.get()
+ )
+ self.config.auto_announce_interval_seconds.set(
+ auto_announce_interval_seconds,
+ )
# enable or disable auto announce based on interval
if auto_announce_interval_seconds > 0:
@@ -14964,13 +14980,17 @@ class ReticulumMeshChat:
self.config.auto_send_failed_messages_to_propagation_node.set(value)
if "lxmf_delivery_transfer_limit_in_bytes" in data:
- value = int(data["lxmf_delivery_transfer_limit_in_bytes"])
+ value = self._coerce_int(data["lxmf_delivery_transfer_limit_in_bytes"])
+ if value is None:
+ value = self.config.lxmf_delivery_transfer_limit_in_bytes.get()
value = max(1000, min(value, 1000 * 1000 * 1000))
self.config.lxmf_delivery_transfer_limit_in_bytes.set(value)
self.message_router.delivery_per_transfer_limit = value / 1000
if "lxmf_propagation_transfer_limit_in_bytes" in data:
- value = int(data["lxmf_propagation_transfer_limit_in_bytes"])
+ value = self._coerce_int(data["lxmf_propagation_transfer_limit_in_bytes"])
+ if value is None:
+ value = self.config.lxmf_propagation_transfer_limit_in_bytes.get()
value = max(1000, min(value, 1000 * 1000 * 100))
self.config.lxmf_propagation_transfer_limit_in_bytes.set(value)
self.message_router.propagation_per_transfer_limit = value / 1000
@@ -14978,7 +14998,9 @@ class ReticulumMeshChat:
self.message_router.announce_propagation_node()
if "lxmf_propagation_sync_limit_in_bytes" in data:
- value = int(data["lxmf_propagation_sync_limit_in_bytes"])
+ value = self._coerce_int(data["lxmf_propagation_sync_limit_in_bytes"])
+ if value is None:
+ value = self.config.lxmf_propagation_sync_limit_in_bytes.get()
value = max(1000, min(value, 1000 * 1000 * 500))
self.config.lxmf_propagation_sync_limit_in_bytes.set(value)
self.message_router.propagation_per_sync_limit = value / 1000
@@ -15027,7 +15049,9 @@ class ReticulumMeshChat:
# update inbound stamp cost (for direct delivery messages)
if "lxmf_inbound_stamp_cost" in data:
- value = int(data["lxmf_inbound_stamp_cost"])
+ value = self._coerce_int(data["lxmf_inbound_stamp_cost"])
+ if value is None:
+ value = self.config.lxmf_inbound_stamp_cost.get()
# 0 disables inbound stamps; otherwise clamp to 1-254 (LXMF/LXMRouter)
if value < 0:
value = 0
@@ -15054,7 +15078,9 @@ class ReticulumMeshChat:
# update propagation node stamp cost (for messages propagated through your node)
if "lxmf_propagation_node_stamp_cost" in data:
- value = int(data["lxmf_propagation_node_stamp_cost"])
+ value = self._coerce_int(data["lxmf_propagation_node_stamp_cost"])
+ if value is None:
+ value = self.config.lxmf_propagation_node_stamp_cost.get()
# validate stamp cost (must be at least 13, per LXMF minimum)
if value < 13:
value = 13
@@ -15069,9 +15095,11 @@ class ReticulumMeshChat:
# update auto sync interval
if "lxmf_preferred_propagation_node_auto_sync_interval_seconds" in data:
- value = int(
+ value = self._coerce_int(
data["lxmf_preferred_propagation_node_auto_sync_interval_seconds"],
)
+ if value is None:
+ value = self.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get()
self.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.set(
value,
)
@@ -15113,14 +15141,14 @@ class ReticulumMeshChat:
)
if "page_archiver_max_versions" in data:
- self.config.page_archiver_max_versions.set(
- int(data["page_archiver_max_versions"]),
- )
+ value = self._coerce_int(data["page_archiver_max_versions"])
+ if value is not None:
+ self.config.page_archiver_max_versions.set(value)
if "archives_max_storage_gb" in data:
- self.config.archives_max_storage_gb.set(
- int(data["archives_max_storage_gb"]),
- )
+ value = self._coerce_int(data["archives_max_storage_gb"])
+ if value is not None:
+ self.config.archives_max_storage_gb.set(value)
if "backup_max_count" in data:
try:
@@ -15526,26 +15554,32 @@ class ReticulumMeshChat:
self.config.voicemail_greeting.set(data["voicemail_greeting"])
if "voicemail_auto_answer_delay_seconds" in data:
- self.config.voicemail_auto_answer_delay_seconds.set(
- int(data["voicemail_auto_answer_delay_seconds"]),
- )
+ value = self._coerce_int(data["voicemail_auto_answer_delay_seconds"])
+ if value is not None:
+ self.config.voicemail_auto_answer_delay_seconds.set(value)
if "voicemail_max_recording_seconds" in data:
- self.config.voicemail_max_recording_seconds.set(
- int(data["voicemail_max_recording_seconds"]),
- )
+ value = self._coerce_int(data["voicemail_max_recording_seconds"])
+ if value is not None:
+ self.config.voicemail_max_recording_seconds.set(value)
if "voicemail_tts_speed" in data:
- self.config.voicemail_tts_speed.set(int(data["voicemail_tts_speed"]))
+ value = self._coerce_int(data["voicemail_tts_speed"])
+ if value is not None:
+ self.config.voicemail_tts_speed.set(value)
if "voicemail_tts_pitch" in data:
- self.config.voicemail_tts_pitch.set(int(data["voicemail_tts_pitch"]))
+ value = self._coerce_int(data["voicemail_tts_pitch"])
+ if value is not None:
+ self.config.voicemail_tts_pitch.set(value)
if "voicemail_tts_voice" in data:
self.config.voicemail_tts_voice.set(data["voicemail_tts_voice"])
if "voicemail_tts_word_gap" in data:
- self.config.voicemail_tts_word_gap.set(int(data["voicemail_tts_word_gap"]))
+ value = self._coerce_int(data["voicemail_tts_word_gap"])
+ if value is not None:
+ self.config.voicemail_tts_word_gap.set(value)
# update ringtone settings
if "custom_ringtone_enabled" in data:
@@ -15553,9 +15587,13 @@ class ReticulumMeshChat:
self._parse_bool(data["custom_ringtone_enabled"]),
)
if "ringtone_preferred_id" in data:
- self.config.ringtone_preferred_id.set(int(data["ringtone_preferred_id"]))
+ value = self._coerce_int(data["ringtone_preferred_id"])
+ if value is not None:
+ self.config.ringtone_preferred_id.set(value)
if "ringtone_volume" in data:
- self.config.ringtone_volume.set(int(data["ringtone_volume"]))
+ value = self._coerce_int(data["ringtone_volume"])
+ if value is not None:
+ self.config.ringtone_volume.set(value)
if "do_not_disturb_enabled" in data:
self.config.do_not_disturb_enabled.set(
@@ -15604,12 +15642,14 @@ class ReticulumMeshChat:
)
if "telephone_tone_generator_volume" in data:
- self.config.telephone_tone_generator_volume.set(
- int(data["telephone_tone_generator_volume"]),
- )
+ value = self._coerce_int(data["telephone_tone_generator_volume"])
+ if value is not None:
+ self.config.telephone_tone_generator_volume.set(value)
if "telephone_audio_profile_id" in data:
- profile_id = int(data["telephone_audio_profile_id"])
+ profile_id = self._coerce_int(data["telephone_audio_profile_id"])
+ if profile_id is None:
+ profile_id = self.config.telephone_audio_profile_id.get()
self.config.telephone_audio_profile_id.set(profile_id)
if self.telephone_manager and self.telephone_manager.telephone:
await asyncio.to_thread(
diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py
index 6c08a45f..1a499ee4 100644
--- a/tests/backend/test_security_fuzzing.py
+++ b/tests/backend/test_security_fuzzing.py
@@ -374,25 +374,6 @@ def test_telephone_callback_fuzzing(mock_app, caller_id_bytes):
mock_app.telephone_manager.on_telephone_call_ended(mock_identity)
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- data=st.dictionaries(
- keys=st.text(),
- values=st.one_of(
- st.text(),
- st.binary(),
- st.integers(),
- st.floats(),
- st.lists(st.text()),
- st.dictionaries(keys=st.text(), values=st.text()),
- ),
- ),
-)
-def test_message_dao_upsert_fuzzing(mock_app, data):
- """Fuzz MessageDAO.upsert_lxmf_message with varied dictionary data."""
- mock_app.database.messages.upsert_lxmf_message(data)
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
title_bytes=st.binary(min_size=0, max_size=1000),
@@ -443,17 +424,6 @@ def test_voicemail_greeting_fuzzing(mock_app, greeting_text):
pass
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(caller_hash=st.binary(min_size=0, max_size=32))
-def test_voicemail_incoming_call_fuzzing(mock_app, caller_hash):
- """Fuzz voicemail incoming call handling."""
- mock_identity = MagicMock()
- mock_identity.hash = caller_hash
-
- # Should not crash with malformed identity hashes
- mock_app.voicemail_manager.handle_incoming_call(mock_identity)
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
source_hash=st.text(min_size=0, max_size=64),
@@ -589,31 +559,6 @@ def test_lxm_generate_paper_uri_fuzzing(mock_app, dest_hash, content):
loop.close()
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- lon=st.floats(allow_nan=False, allow_infinity=False),
- lat=st.floats(allow_nan=False, allow_infinity=False),
- zoom=st.integers(min_value=-100, max_value=100),
-)
-def test_map_manager_coord_fuzzing(mock_app, lon, lat, zoom):
- """Fuzz coordinate to tile conversion in MapManager."""
- # Should handle invalid coordinates gracefully
- mock_app.map_manager._lonlat_to_tile(lon, lat, zoom)
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- text=st.text(),
- source_lang=st.text(min_size=0, max_size=10),
- target_lang=st.text(min_size=0, max_size=10),
-)
-def test_translator_handler_fuzzing(mock_app, text, source_lang, target_lang):
- """Fuzz the TranslatorHandler translate_text method."""
- mock_app.translator_handler.has_requests = False
- mock_app.translator_handler.has_argos = False
- mock_app.translator_handler.translate_text(text, source_lang, target_lang)
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(dest_hash=st.text(), icon_name=st.text(), fg_color=st.text(), bg_color=st.text())
def test_update_lxmf_user_icon_fuzzing(
@@ -1153,16 +1098,6 @@ def test_map_tile_coordinates_fuzzing(mock_app, z, x, y):
pass
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- filename=st.text(min_size=0, max_size=500),
-)
-def test_mbtiles_filename_fuzzing(mock_app, filename):
- """Fuzz MBTiles filename handling."""
- mock_app.map_manager.delete_mbtiles(filename)
- mock_app.map_manager.get_connection(filename)
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
destination_hash=st.text(min_size=0, max_size=100),
@@ -1181,28 +1116,6 @@ def test_archive_page_content_fuzzing(mock_app, destination_hash, page_path, con
mock_app.archiver_manager.get_archived_page_versions(destination_hash, page_path)
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- ids=st.lists(
- st.one_of(st.integers(), st.text(), st.floats()),
- min_size=0,
- max_size=100,
- ),
-)
-def test_delete_archived_pages_ids_fuzzing(mock_app, ids):
- """Fuzz SQL injection in delete_archived_pages."""
- mock_app.database.misc.delete_archived_pages(ids=ids)
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- query=st.text(min_size=0, max_size=500),
-)
-def test_archived_pages_query_sql_injection_fuzzing(mock_app, query):
- """Fuzz SQL injection in archived_pages search."""
- mock_app.database.misc.get_archived_pages_paginated(query=query)
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
file_path=st.text(min_size=0, max_size=1000),
@@ -1507,13 +1420,6 @@ def test_nomadnet_page_download_fuzzing(
loop.close()
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(archive_id=st.one_of(st.integers(), st.text(), st.floats(), st.none()))
-def test_get_archived_page_by_id_fuzzing(mock_app, archive_id):
- """Fuzz archived page lookup by id (SQL injection, type confusion)."""
- mock_app.database.misc.get_archived_page_by_id(archive_id)
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
forward_to_hash=st.text(min_size=0, max_size=100),
@@ -1625,44 +1531,6 @@ def test_keyboard_shortcuts_set_fuzzing(mock_app, action, keys):
loop.close()
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(message_hash=st.text(min_size=0, max_size=200))
-def test_messages_get_by_hash_fuzzing(mock_app, message_hash):
- """Fuzz message lookup by hash (SQL, type confusion)."""
- mock_app.database.messages.get_lxmf_message_by_hash(message_hash)
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(message_hash=st.text(min_size=0, max_size=200))
-def test_messages_delete_by_hash_fuzzing(mock_app, message_hash):
- """Fuzz single message delete by hash."""
- try:
- mock_app.database.messages.delete_lxmf_message_by_hash(message_hash)
- except Exception:
- pass
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- message_hashes=st.lists(st.text(min_size=0, max_size=100), min_size=0, max_size=50),
-)
-def test_messages_delete_by_hashes_fuzzing(mock_app, message_hashes):
- """Fuzz bulk message delete by hashes."""
- try:
- mock_app.database.messages.delete_lxmf_messages_by_hashes(message_hashes)
- except Exception:
- pass
-
-
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- table_name=st.text(min_size=0, max_size=100),
-)
-def test_sql_table_name_injection_fuzzing(mock_app, table_name):
- """Fuzz SQL table name injection."""
- mock_app.database.provider.execute(f"PRAGMA table_info({table_name})")
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
fields_json=st.text(min_size=0, max_size=10000),
@@ -1800,17 +1668,6 @@ def test_map_export_parameters_fuzzing(mock_app, bbox, min_zoom, max_zoom, name)
)
-@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
-@given(
- mbtiles_path=st.text(min_size=0, max_size=1000),
-)
-def test_mbtiles_metadata_parsing_fuzzing(mock_app, mbtiles_path):
- """Fuzz MBTiles metadata parsing."""
- mock_app.map_manager.get_metadata()
- if os.path.exists(mbtiles_path):
- mock_app.map_manager.get_connection(mbtiles_path)
-
-
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
audio_frame=st.one_of(
diff --git a/tests/backend/test_telemetry_integration.py b/tests/backend/test_telemetry_integration.py
index e1efc27e..8f7f8ed1 100644
--- a/tests/backend/test_telemetry_integration.py
+++ b/tests/backend/test_telemetry_integration.py
@@ -165,10 +165,5 @@ async def test_telemetry_request_no_location_does_not_call_handler(mock_app):
mock_app.handle_telemetry_request.assert_not_called()
-@pytest.mark.asyncio
-async def test_tracking_toggle_endpoint(mock_app):
- # Mock database responses
- mock_app.database.telemetry.is_tracking.return_value = False
-
# We can't easily test the web endpoint here without more setup,
# but we can test the logic it calls if it was refactored into a method.
diff --git a/tests/backend/test_translator_handler_extended.py b/tests/backend/test_translator_handler_extended.py
index 0a33a319..501e0217 100644
--- a/tests/backend/test_translator_handler_extended.py
+++ b/tests/backend/test_translator_handler_extended.py
@@ -109,8 +109,6 @@ def test_translate_argos_cli(mock_run):
assert result["translated_text"] == "Hola"
-def test_detect_language_simple():
- TranslatorHandler()
# _detect_language is private
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────